package com.xm.listener;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.MessageProperties;
import com.xm.model.Msg;
import com.xm.util.old.JacksonUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

@Component
@Slf4j
public class AckReceiver {
    private final AtomicInteger number = new AtomicInteger();
    private final ConcurrentHashMap<String, Object> concurrentHashMap = new ConcurrentHashMap<>();

    @RabbitHandler
    @RabbitListener(queues = "directQueueAck")
    public void process(String msgStr, Message amqpMsg, Channel channel) throws IOException {
        Msg msg = null;
        int i = 0;

        long deliveryTag = amqpMsg.getMessageProperties().getDeliveryTag();

        try {
            i = number.addAndGet(1);

            msg = JacksonUtil.string2Obj(msgStr, Msg.class);
            if(msg == null){
                log.error("ackReceiver(process)消费者收到消息[{}]异常，消费消息格式异常：{}，deliveryTag={}", i, msg, deliveryTag);
                throw new RuntimeException("监听消息消费异常：消费消息格式异常！");
            }

//            String message = new String(amqpMsg.getBody(), StandardCharsets.UTF_8);
            //测试消息逐条消费
            if(5 == i
                    || 6 == i || 7 == i
                    || 8 == i//模拟消息5重试三次依旧失败，此时应将异常消息写入数据库
                    || 9 == i){//模拟部分异常消息，重试后正常
                Thread.sleep(2000L);
                throw new RuntimeException("监听消息消费异常：5的消息消费异常！");//默认异常消息重回队头，控制台会一直提示‘Caused by: java.lang.RuntimeException: 5的消息消费异常！’，且会阻塞线程
            }else{
                Thread.sleep(1000L);
            }

            //人为调用方法确认
            //basicAck：参数1-消息唯一标识；参数2-false时，仅处理当前消息
            channel.basicAck(deliveryTag, false);
            log.info("ackReceiver(process)消费者收到消息[{}]：{}，deliveryTag={}", i, msg, deliveryTag);
        } catch (Exception e) {
            if(msg != null){
                //处理一：重试三次
                Integer retryCount;
                String msgId = msg.getMsgId();

                if(!concurrentHashMap.containsKey(msgId)){
                    retryCount = 1;
                }else{
                    retryCount = (Integer) concurrentHashMap.get(msgId);
                    retryCount ++;
                }
                concurrentHashMap.put(msgId, retryCount);

                if(retryCount <= 3){
                    //处理一：未超过重试次数上限，重新发布至队头，并立刻重试
                    channel.basicReject(deliveryTag, true);
                    log.error("ackReceiver(process)消费者收到消息[{}]异常，重新发布至队头，重试{}次：{}", i, retryCount, msg);
                }else{
                    //处理二：超过重试次数上限，重新发布至队尾，并写入数据库
                    channel.basicAck(deliveryTag, false);// 一定要先删除 再添加，不然当前线程会阻塞
                    //捕获异常，将消息发回指定交换机 指定路由，并放置队尾
                    channel.basicPublish(amqpMsg.getMessageProperties().getReceivedExchange(),
                            amqpMsg.getMessageProperties().getReceivedRoutingKey(),
                            MessageProperties.PERSISTENT_TEXT_PLAIN,
                            amqpMsg.getBody());
                    log.error("ackReceiver(process)消费者收到消息[{}]异常，超过重试次数上限，重新发布至队尾：{}", i, msg);

                    //写入数据库（模拟）
                    log.info("ackReceiver(process)消费者收到消息[{}]异常，请尽快处理。正在向数据库写入异常消息。。msg={}", i, msg);
                    //...
                }
            }
        } finally {
        }
    }

    @RabbitHandler
    @RabbitListener(queues = "directQueueAck")
    public void process2(Msg msg, Message amqpMsg, Channel channel) throws IOException {
        long deliveryTag = amqpMsg.getMessageProperties().getDeliveryTag();
        int i = number.addAndGet(1);
        log.info("ackReceiver(process2)消费者收到消息[{}]，msg={}", i, msg);
        channel.basicAck(deliveryTag, false);
    }

    @RabbitHandler
    @RabbitListener(queues = "directQueueAck11")
    public void process3(String msgStr, Message amqpMsg, Channel channel) throws IOException {
        long deliveryTag = amqpMsg.getMessageProperties().getDeliveryTag();
        int i = number.addAndGet(1);
        Msg msg = JacksonUtil.string2Obj(msgStr, Msg.class);
        log.info("ackReceiver(process3)消费者收到消息[{}]，msg={}", i, msg);
        assert msg != null;
        if(msg.getType() == 0){
            channel.basicAck(deliveryTag, false);
        }else if(msg.getType() == 1){
            channel.basicReject(deliveryTag, false);
        }else if(msg.getType() == 2){
            channel.basicNack(deliveryTag, false, false);
        }else if(msg.getType() == 3){
            //准发消息（私信队列）
            channel.basicAck(deliveryTag, false);
            AMQP.BasicProperties props = new AMQP.BasicProperties().builder()
                    .contentType("application/json")
                    .build();
            channel.basicPublish("directExchangeAckNew","directRoutingKeyAckNew", props, msgStr.getBytes(StandardCharsets.UTF_8));
            channel.basicPublish("directExchangeAckNew","directRoutingKeyAckNew", props, amqpMsg.getBody());
        }else{
            throw new RuntimeException("消息类型错误");
        }
    }

}
